﻿#include "ssAsioNetWork.h"
//#include "../../StarSeekerBaseLib/ssAssert.h"
#include "log/ssLogger.h"
#include "message/ssMsgIdDef.h"
#include "message/ssNetMsgHead.h"
//#include "../../StarSeekerBaseLib/thread/ssThread.h"
//#include "../../StarSeekerBaseLib/time/ssTime.h"

namespace StarSeeker
{

ssAsioNetWork::ssAsioNetWork(void)
	:m_acceptor(m_ioservice),m_sessionCount(100),m_threadCount(1),
	m_recvBuffSize(SS_ASIO_PACK_SIZE), m_isClient(false), m_exiting(false),m_sendEventCount(0)
{
	m_sessionID.Init(1, 400000000);
}

ssAsioNetWork::~ssAsioNetWork(void)
{
	StopNetWork();
	if (!m_isClient)
		m_acceptor.close();
}

void ssAsioNetWork::InitParam(u32 sessionCount, u32 threadCount, u32 recvBuffSize)
{
	m_sessionCount = sessionCount;
	m_threadCount = threadCount;
	m_recvBuffSize = recvBuffSize;
}

bool ssAsioNetWork::StartNetWork(const char *ip, const u16 port, bool isClient)
{
	if (m_threads.empty())
	{
		m_exiting = false;
		m_isClient = isClient;
		boost::asio::ip::address addrip;
		addrip = addrip.from_string(ip);
		m_addr.address(addrip);
		m_addr.port(port);
		bool isv4 = addrip.is_v4();
		if (m_isClient)
		{
			ConnectToServer();
		}
		else
		{
			try
			{
				m_acceptor.open(isv4 ? boost::asio::ip::tcp::v4() : boost::asio::ip::tcp::v6());
				//m_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
				m_acceptor.bind(m_addr);
				m_acceptor.listen();
			}
			catch (boost::system::system_error& e)
			{
				SS_LOG->ErrorEx("ssAsioNetWork::StartNetWork error:%s", e.what());
				return false;
			}
			catch (...)
			{
				SS_LOG->Error("ssAsioNetWork::StartNetWork error");
				return false;
			}
			DoAccept();
			SS_LOG->PrintEx("start listen %s:%d", ip, port);
		}
		if (m_exiting)
			return false;
		for (u32 i=0; i<m_threadCount; ++i)
		{
			threadPtr pThread(new boost::thread(boost::bind(&ssIOService::RunService, &m_ioservice)));
			m_threads.push_back(pThread);
		}
	}
	return false;
}

void ssAsioNetWork::StopNetWork()
{
	m_exiting = true;
    m_ioservice.stop();
	for (u32 i = 0; i<m_threads.size(); ++i)
	{
		if (m_threads[i])
		{
			m_threads[i]->join();
			//SS_SAFE_DELETE(m_threads[i])
		}
	}
	m_threads.clear();
	ClearSession();
}

void ssAsioNetWork::DoAccept()
{
	boost::shared_ptr<boost::asio::ip::tcp::socket> sk2(new boost::asio::ip::tcp::socket(m_ioservice));
	m_acceptor.async_accept(*sk2, boost::bind(&ssAsioNetWork::HandleAccept, this, sk2, boost::asio::placeholders::error));
}

void ssAsioNetWork::DoReceive(ssAsioSessionPtr& ptr)
{
	if (ptr->m_recvBuf.ResidueSize() < 100)
	{
		SS_LOG->ErrorEx("%d recvBuf residueSize<100:%u port:%d", ptr->m_sessionID, (u32)ptr->m_recvBuf.ResidueSize(), m_addr.port());
	}
	ptr->m_socket->async_read_some(boost::asio::buffer(ptr->m_recvBuf.CurrentBuf(), ptr->m_recvBuf.ResidueSize()), boost::bind(&ssAsioNetWork::HandleAfterRead, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred, ptr->m_sessionID));
}

void ssAsioNetWork::HandleAccept(boost::shared_ptr<boost::asio::ip::tcp::socket> s, const boost::system::error_code& error)
{
	if (s)
	{
		if (!error)
		{
			if (m_sessionMap.size() > m_sessionCount)
			{
				boost::system::error_code ec1, ec2;
				s->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec1);
				s->close(ec2);
			}
			else
			{
				//!!需要对象池
				ssAsioSessionPtr pSession (new ssAsioSession());
				pSession->Init(m_sessionID.GetID(), s, m_recvBuffSize, this);
				m_sessionMap.insert(std::pair<u32, ssAsioSessionPtr>(pSession->GetID(), pSession));
				DoReceive(pSession);
				OnNewSession(pSession->GetID());
			}
		}
		else
		{
			boost::system::error_code ec1, ec2;
			s->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec1);
			s->close(ec2);
		}
	}
	DoAccept();
}

void ssAsioNetWork::DoCloseSocket(u32 sessionId, asioClose_e errTypr, i32 err)
{
	m_ioservice.post(boost::bind(&ssAsioNetWork::HandleCloseSocket, this, sessionId, errTypr));
}

void ssAsioNetWork::DoSendData(u32 sessionId, char* buf, u32 len)
{
	m_ioservice.post(boost::bind(&ssAsioNetWork::HandleSendData, this, sessionId, boost::shared_array<char>(buf), len));
}

void ssAsioNetWork::DoSendDataMulti(std::vector<u32>& sessionId, char* buf, u32 len)
{
	m_ioservice.post(boost::bind(&ssAsioNetWork::HandleSendMulti, this, sessionId, boost::shared_array<char>(buf), len));
}

void ssAsioNetWork::DoSendDataAll(char* buf, u32 len)
{
	m_ioservice.post(boost::bind(&ssAsioNetWork::HandleSendAll, this, boost::shared_array<char>(buf), len));
}

void ssAsioNetWork::Reconnect()
{
	if (m_isClient && !IsConnect())
	{
		ConnectToServer();
	}
}

StarSeeker::u32 ssAsioNetWork::GetClientID()
{
	if (m_isClient && !m_sessionMap.empty())
	{
		return m_sessionMap.begin()->second->GetID();
	}
	return 0;
}

StarSeeker::u32 ssAsioNetWork::GetSessionCount()
{
	return (u32)m_sessionMap.size();
}


StarSeeker::lint ssAsioNetWork::GetSendEventCount()
{
	return m_sendEventCount;
}

void ssAsioNetWork::OnSessionMessage(const char* ip, u32 sessionID, u32 msgID, char* data, ssNetMsgHead* head)
{
	if (data)
	{
		delete[] data;
	}
}

void ssAsioNetWork::HandleCloseSocket(u32 sessionId, asioClose_e origin)
{
	ssAsioSessionPtr ps = FindSession(sessionId);
	if (ps)
	{
		try
		{
			boost::system::error_code ec1, ec2;
			ps->GetSocket()->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec1);
			ps->GetSocket()->close(ec2);
		}
		catch (boost::system::system_error& e)
		{
			SS_LOG->ErrorEx("HandleCloseSocket %d by%d %s", sessionId, origin, e.what());
		}
		catch (boost::system::error_code& e)
		{
			SS_LOG->ErrorEx("HandleCloseSocket %d by%d %s", sessionId, origin, e.message().c_str());
		}
		catch (...)
		{
		}

		m_sessionMap.erase(sessionId);
		//delete ps;
		OnSessionClose(sessionId);
	}
	else
	{
		SS_LOG->ErrorEx("HandleCloseSocket sessionId not found:%d by%d", sessionId, origin);
	}
}

void ssAsioNetWork::HandleSendData(u32 sessionId, boost::shared_array<char> buf, u32 len)
{
	if (buf && len)
	{
		if (len < MSG_MAX_SIZE)
		{
			ssAsioSessionPtr ps = FindSession(sessionId);
			if (ps)
			{
				m_sendEventCount++;
				boost::asio::async_write(*ps->m_socket, boost::asio::buffer(buf.get(), len),
				boost::bind(&ssAsioNetWork::HandleAfterWrite, this, _1, _2, ps->m_sessionID, buf));
			}
		}
		else
		{
			SS_LOG->ErrorEx("HandleSendData data too long:%d", sessionId);
		}
	}
}

void ssAsioNetWork::HandleSendMulti(std::vector<u32>& sessionId, boost::shared_array<char> buf, u32 len)
{
	if (buf && len)
	{
		if (len < MSG_MAX_SIZE)
		{
			for (u32 i = 0; i < sessionId.size(); ++i)
			{
				ssAsioSessionPtr ps = FindSession(sessionId[i]);
				if (ps)
				{
					m_sendEventCount++;
					boost::asio::async_write(*ps->m_socket, boost::asio::buffer(buf.get(), len),
						boost::bind(&ssAsioNetWork::HandleAfterWrite, this, _1, _2, ps->m_sessionID, buf));
				}
			}
		}
		else
		{
			SS_LOG->Error("HandleSendMulti data too long");
		}
	}
}

void ssAsioNetWork::HandleSendAll(boost::shared_array<char> buf, u32 len)
{
	if (buf && len)
	{
		if (len < MSG_MAX_SIZE)
		{
			for (std::map<u32, ssAsioSessionPtr>::iterator i = m_sessionMap.begin(); i != m_sessionMap.end(); ++i)
			{
				m_sendEventCount++;
				boost::asio::async_write(*i->second->m_socket, boost::asio::buffer(buf.get(), len),
				boost::bind(&ssAsioNetWork::HandleAfterWrite, this, _1, _2, i->second->m_sessionID, buf));
			}
		}
		else
		{
			SS_LOG->Error("HandleSendAll data too long");
		}
	}
}

void ssAsioNetWork::HandleAfterRead(const boost::system::error_code & error, size_t bytes, u32 sessionId)
{
	if (!error)
	{
		ssAsioSessionPtr ps = FindSession(sessionId);
		if (ps)
		{
			ps->m_recvBuf.AddSize(bytes);
			ps->OnReceive();
			DoReceive(ps);
		}
	}
	else
	{
		std::string errstr("HandleAfterRead %u");
		errstr += error.message();
		SS_LOG->ErrorEx(errstr.c_str(), sessionId);
		HandleCloseSocket(sessionId, CLOSE_BY_READ);
	}
}

void ssAsioNetWork::HandleAfterWrite(const boost::system::error_code & error, size_t bytes, u32 sessionId, boost::shared_array<char> buf)
{
	m_sendEventCount--;
	if (error)
	{
		std::string errstr("HandleAfterWrite %u");
		errstr += error.message();
		SS_LOG->ErrorEx(errstr.c_str(), sessionId);
		HandleCloseSocket(sessionId, CLOSE_BY_WRITE);
	}
}

ssAsioSessionPtr ssAsioNetWork::FindSession(u32 id)
{
	ssAsioSessionPtr ret;
	std::map<u32, ssAsioSessionPtr>::iterator itr = m_sessionMap.find(id);
	if (itr != m_sessionMap.end())
	{
		ret = itr->second;
	}
	return ret;
}

bool ssAsioNetWork::IsConnect()
{
	if (m_isClient)
	{
		return !m_sessionMap.empty();
	}
	return true;
}

void ssAsioNetWork::ClearSession()
{
	std::map<u32, ssAsioSessionPtr>::iterator itr;
	for (itr = m_sessionMap.begin(); itr != m_sessionMap.end(); ++itr)
	{
		boost::system::error_code ec1, ec2;
		itr->second->GetSocket()->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec1);
		itr->second->GetSocket()->close(ec2);
		//delete itr->second;
	}
	m_sessionMap.clear();
}

bool ssAsioNetWork::ConnectToServer()
{
	boost::system::error_code err;
	boost::shared_ptr<boost::asio::ip::tcp::socket> sk(new boost::asio::ip::tcp::socket(m_ioservice));
	do
	{
		SS_LOG->PrintEx("connect to %s:%d", m_addr.address().to_string().c_str(), m_addr.port());
		sk->connect(m_addr, err);
		if (!err)
		{
			//!!需要对象池
			ssAsioSessionPtr pSession(new ssAsioSession());
			pSession->Init(1, sk, m_recvBuffSize, this);
			m_sessionMap.insert(std::pair<u32, ssAsioSessionPtr>(pSession->GetID(), pSession));
			DoReceive(pSession);
			OnNewSession(pSession->GetID());
			return true;
		}

		ssThreadBase::Sleep(10000);
	} while (err && !m_exiting);
	return false;
}

}
